/*
 * 根据Apache软件基金会(ASF)的许可证条款授权。
 * 详情请见NOTICE文件和LICENSE文件。
 */

package org.apache.flink.training.solutions.hourlytips;

import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.training.exercises.common.datatypes.TaxiFare;
import org.apache.flink.training.exercises.common.sources.TaxiFareGenerator;
import org.apache.flink.util.Collector;

/**
 * Flink培训课程中"每小时小费统计"练习的Java参考实现。
 * 任务目标：
 * 1. 首先计算每个司机每小时收取的小费总额
 * 2. 从上述数据流中找出每小时最高的小费总额
 *
 * 类比说明：想象一个出租车公司实时监控系统，每小时内统计每个司机的小费收入，然后找出每小时的"小费冠军"
 */
public class HourlyTipsSolution {

    private final SourceFunction<TaxiFare> source; // 数据源（出租車费用数据生成器）
    private final SinkFunction<Tuple3<Long, Long, Float>> sink; // 数据输出端（打印到控制台）

    /** 使用指定的数据源和数据输出端创建任务 */
    public HourlyTipsSolution(
            SourceFunction<TaxiFare> source, SinkFunction<Tuple3<Long, Long, Float>> sink) {
        this.source = source;
        this.sink = sink;
    }

    /**
     * 程序主入口
     * @throws Exception 可能出现的执行异常
     */
    public static void main(String[] args) throws Exception {
        // 创建任务实例：使用出租车费用生成器作为数据源，控制台打印作为输出
        HourlyTipsSolution job =
                new HourlyTipsSolution(new TaxiFareGenerator(), new PrintSinkFunction<>());

        job.execute(); // 执行任务
    }

    /**
     * 创建并执行每小时小费统计流水线
     * @return 任务执行结果
     * @throws Exception 可能出现的执行异常
     */
    public JobExecutionResult execute() throws Exception {
        // 设置流处理执行环境（类似建立一条生产线的基础框架）
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 启动数据生成器并设置水印（用于处理事件时间）
        DataStream<TaxiFare> fares =
                env.addSource(source)
                        .assignTimestampsAndWatermarks(
                                // 出租车费用数据是按时间顺序生成的（类似按时间顺序排列的发票存根）
                                WatermarkStrategy.<TaxiFare>forMonotonousTimestamps()
                                        .withTimestampAssigner(
                                                (fare, t) -> fare.getEventTimeMillis())); // 提取事件时间戳

        // 计算每个司机每小时的小费总额（类似先按司机分组，再按小时汇总）
        DataStream<Tuple3<Long, Long, Float>> hourlyTips =
                fares.keyBy((TaxiFare fare) -> fare.driverId) // 按司机ID分组（类似将发票按司机分类）
                        .window(TumblingEventTimeWindows.of(Time.hours(1))) // 1小时滚动窗口（类似每小时结算一次）
                        .process(new AddTips()); // 处理窗口数据（自定义聚合逻辑）

        // 找出每小时最高的小费总额（从所有司机中找出每小时的冠军）
        DataStream<Tuple3<Long, Long, Float>> hourlyMax =
                hourlyTips.windowAll(TumblingEventTimeWindows.of(Time.hours(1))).maxBy(2); // 全局窗口取最大值

        /*
         * 可以尝试用下面的替代方案（已注释），比较两者差异：
         * 这种方案是先按窗口时间分组，然后取最大值，与全局窗口方案的区别在于：
         * - 窗口All操作：将所有数据汇集到一个任务槽进行全局计算（类似总经理收集所有部门报告后评选最佳）
         * - 按键分区后取最大值：并行计算，每个时间窗口的数据分散在不同任务槽（类似各部门先内部评选，再上报）
         */
        // DataStream<Tuple3<Long, Long, Float>> hourlyMax = hourlyTips.keyBy(t -> t.f0).maxBy(2);

        hourlyMax.addSink(sink); // 将结果输出到控制台

        return env.execute("Hourly Tips"); // 启动任务执行（类似启动生产线）
    }

    /*
     * 自定义窗口处理函数：将小费累加后包装成(窗口结束时间, 司机ID, 小费总额)三元组
     * 类比：会计人员汇总每个司机每小时的小费收入，并记录结算时间和司机编号
     */
    public static class AddTips
            extends ProcessWindowFunction<TaxiFare, Tuple3<Long, Long, Float>, Long, TimeWindow> {

        @Override
        public void process(
                Long key, // 司机ID
                Context context, // 窗口上下文
                Iterable<TaxiFare> fares, // 窗口内的费用数据集合
                Collector<Tuple3<Long, Long, Float>> out) { // 结果收集器

            float sumOfTips = 0F;
            for (TaxiFare f : fares) {
                sumOfTips += f.tip; // 累加小费（类似会计逐笔累加发票金额）
            }
            // 输出格式：(窗口结束时间, 司机ID, 小费总额)
            out.collect(Tuple3.of(context.window().getEnd(), key, sumOfTips));
        }
    }
}